﻿/*                              
   Copyright 2011 Nils Kopal, Christopher Krüger, Simone Sauer, Dennis Nolte, Uni Duisburg-Essen

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Numerics;
using System.Timers;
using Sharebert.LinkLayerLibrary;
using Sharebert.LinkLayerLibrary.API;
using Sharebert.OverlayLibrary;
using Sharebert.OverlayLibrary.API;
using Sharebert.DHTLibrary.API;
using Sharebert.DHTLibrary.Messages;
using log4net;

namespace Sharebert.DHTLibrary
{
    public class DistributedHashtable : IDHT
    {
	private static readonly ILog logger = LogManager.GetLogger(typeof(DistributedHashtable));
        private IOverlay _overlay;
        private ThreadSafeDictionary<BigInteger, DHTEntry> _dictionary;
        private BigInteger _startIndex = BigInteger.Zero;
        private BigInteger _endIndex = BigInteger.Zero;
        private int _housekeepingInterval = 60000;

        public DistributedHashtable(IOverlay overlay, int housekeepingInterval = 60000)
        {
            _overlay = overlay;
            _overlay.OverlayMessageReceived += new EventHandler<OverlayMessageReceivedEventArgs>(_overlay_OverlayMessageReceived);
            _dictionary = new ThreadSafeDictionary<BigInteger, DHTEntry>();
            _endIndex = overlay.LocalPeer.ID.BigIntegerValue;
            _housekeepingInterval = housekeepingInterval;

            var timer = new System.Timers.Timer();
            timer.Interval = housekeepingInterval;
            timer.Elapsed += new ElapsedEventHandler(housekeeping_timer_elapsed);
            timer.Start();
        }

        private void housekeeping_timer_elapsed(object sender, ElapsedEventArgs e)
        {
            var keys = _dictionary.Keys;

            foreach (BigInteger i in keys)
            {
                if (DateTime.Now > _dictionary[i].Timestamp.AddMilliseconds(_housekeepingInterval))
                {
                    logger.Debug("Remove keyhash=" + Tools.Hash42(_dictionary[i].data) + " from DHT because its timestamp is too old");
                    _dictionary.Remove(i);
                }
            }
        }

        private void CheckPredecessor()
        {
            _endIndex = _overlay.LocalPeer.ID.BigIntegerValue;
            if (_overlay.Predecessors.Count > 0)
            {
                _startIndex = _overlay.Predecessors[0].RemotePeer.ID.BigIntegerValue;
            }
            else
            {
                _startIndex = _overlay.LocalPeer.ID.BigIntegerValue;
            }
        }

        private void _overlay_OverlayMessageReceived(object sender, OverlayMessageReceivedEventArgs args)
        {
            
            try
            {
                CheckPredecessor();

                if (args.Message.MessageType.Equals(MessageTypes.STORE_MESSAGE))
                {
                    logger.Debug("Retrieved a STORE_MESSAGE from " + args.Message.SenderID);
                    var storeMessage = (StoreMessage)Serialisation.Deserialize(args.Payload, typeof(StoreMessage));
                    Store(storeMessage);
                }
                else if (args.Message.MessageType.Equals(MessageTypes.STORE_SUCCEEDED_MESSAGE))
                {
                    var storeSucceededMessage = (StoreSucceededMessage)Serialisation.Deserialize(args.Payload, typeof(StoreSucceededMessage));
                    logger.Debug("Retrieved a STORE_SUCCEEDED_MESSAGE from " + args.Message.SenderID + " for " + Tools.Hash42(storeSucceededMessage.Key));                    
                    if (StoreSucceeded != null)
                    {
                        StoreSucceeded.Invoke(this, new StoreEventArgs(storeSucceededMessage.Key, storeSucceededMessage.Value));
                    }
                   
                }
                else if (args.Message.MessageType.Equals(MessageTypes.RETRIEVE_MESSAGE))
                {
                    logger.Debug("Retrieved a RETRIEVE_MESSAGE from " + args.Message.SenderID);
                    var retrieveMessage = (RetrieveMessage)Serialisation.Deserialize(args.Payload, typeof(RetrieveMessage));
                    Retrieve(retrieveMessage);
                }
                else if (args.Message.MessageType.Equals(MessageTypes.RETRIEVE_RESPONSE_MESSAGE))
                {
                    logger.Debug("Retrieved a RETRIEVE_RESPONSE_MESSAGE from " + args.Message.SenderID);
                    var retrieveMessage = (RetrieveResponseMessage)Serialisation.Deserialize(args.Payload, typeof(RetrieveResponseMessage));                   
                    if (RetrieveSucceeded != null)
                    {
                        RetrieveSucceeded.Invoke(this, new RetrieveEventArgs(retrieveMessage.Key, retrieveMessage.Value));
                    }                    
                }
                else if (args.Message.MessageType.Equals(MessageTypes.NOT_FOUND_MESSAGE))
                {
                    logger.Debug("Retrieved a NOT_FOUND_MESSAGE from " + args.Message.SenderID);
                    var notFoundMessage = (NotFoundMessage)Serialisation.Deserialize(args.Payload, typeof(NotFoundMessage));
                    if (KeyNotFound != null)
                    {
                        KeyNotFound.Invoke(this, new RetrieveEventArgs(notFoundMessage.Key, null));
                    }
                }
                else
                {
                    logger.Warn("Retrieved message with unknown message_type: " + args.Message.MessageType);
                }
            }
            catch (Exception ex)
            {
                logger.Error("Exception during _overlay_OverlayMessageReceived.", ex);
            }
        }
        
        #region IDHT Members
        
        /// <summary>
        /// Store a key+value-pair into our dht
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        public void Store(byte[] key, byte[] value)
        {
            try
            {
                var storeMessage = new StoreMessage();
                storeMessage.MessageType = MessageTypes.STORE_MESSAGE;
                storeMessage.SenderID = _overlay.LocalPeer.ID;
                storeMessage.RecipientID = new PeerID(Tools.Hash42(key));
                storeMessage.Key = key;
                storeMessage.Value = value;
                Store(storeMessage);
            }
            catch (Exception ex)
            {
                logger.Error("Exception during local Store.", ex);
            }
        }

        private void Store(StoreMessage storeMessage)
        {
            try
            {
                CheckPredecessor();
                byte[] key = storeMessage.Key;
                byte[] value = storeMessage.Value;
                var i = Tools.Hash42(key);
                if (_startIndex <= _endIndex)
                {
                    if (i > _startIndex && i <= _endIndex || _startIndex == _endIndex)
                    {
                        //we are responsible for this key, so we store the data
                        _dictionary[i] = new DHTEntry() { Timestamp = DateTime.Now, data = value };
                        logger.Debug("Successfully stored keyhash=" + i + " from " + storeMessage.SenderID);
                        if (storeMessage.SenderID == _overlay.LocalPeer.ID)
                        {
                            if (StoreSucceeded != null)
                            {
                                StoreSucceeded.Invoke(this, new StoreEventArgs(key, value));
                            }
                        }
                        else
                        {
                            var storeSucceededMessage = new StoreSucceededMessage();
                            storeSucceededMessage.MessageType = MessageTypes.STORE_SUCCEEDED_MESSAGE;
                            storeSucceededMessage.SenderID = _overlay.LocalPeer.ID;
                            storeSucceededMessage.RecipientID = storeMessage.SenderID;
                            storeSucceededMessage.Key = key;
                            storeSucceededMessage.Value = value;
                            _overlay.Send(storeSucceededMessage);
                        }
                    }
                    else
                    {
                        //we are NOT responsible for this key, so we route it to another peer
                        logger.Debug("We are not responsible for keyhash=" + i + " from " + storeMessage.SenderID);
                        _overlay.Send(storeMessage);
                    }
                }
                else
                {
                    if (i > _startIndex || i <= _endIndex)
                    {
                        //we are responsible for this key, so we store the data
                        _dictionary[i] = new DHTEntry() { Timestamp = DateTime.Now, data = value };
                        logger.Debug("Successfully stored keyhash=" + i + " from " + storeMessage.SenderID);
                        if (storeMessage.SenderID == _overlay.LocalPeer.ID)
                        {
                            if (StoreSucceeded != null)
                            {
                                StoreSucceeded.Invoke(this, new StoreEventArgs(key, value));
                            }
                        }
                        else
                        {
                            var storeSucceededMessage = new StoreSucceededMessage();
                            storeSucceededMessage.MessageType = MessageTypes.STORE_SUCCEEDED_MESSAGE;
                            storeSucceededMessage.SenderID = _overlay.LocalPeer.ID;
                            storeSucceededMessage.RecipientID = storeMessage.SenderID;
                            storeSucceededMessage.Key = key;
                            storeSucceededMessage.Value = value;
                            _overlay.Send(storeSucceededMessage);
                        }
                    }
                    else
                    {
                        //we are NOT responsible for this key, so we route it to another peer
                        logger.Debug("We are not responsible for keyhash=" + i + " from " + storeMessage.SenderID);
                        _overlay.Send(storeMessage);
                    }
                }
            }
            catch (Exception ex)
            {
                logger.Error("Exception during distributed Store.", ex);
            }
        }        

        /// <summary>
        /// Retrieve a value belonging to a key from the dht
        /// </summary>
        /// <param name="key"></param>
        public void Retrieve(byte[] key)
        {
            try
            {
                var retrieveMessage = new RetrieveMessage();
                retrieveMessage.MessageType = MessageTypes.RETRIEVE_MESSAGE;
                retrieveMessage.SenderID = _overlay.LocalPeer.ID;
                retrieveMessage.RecipientID = new PeerID(Tools.Hash42(key));
                retrieveMessage.Key = key;
                Retrieve(retrieveMessage);
            }
            catch (Exception ex)
            {
                logger.Error("Exception during local Retrieve.", ex);
            }
        }

        private void Retrieve(RetrieveMessage retrieveMessage)
        {
            try
            {
                CheckPredecessor();
                byte[] key = retrieveMessage.Key;
                PeerID sender = retrieveMessage.SenderID;
                var i = Tools.Hash42(key);
                if (_startIndex < _endIndex)
                {
                    if (i > _startIndex && i <= _endIndex || _startIndex == _endIndex)
                    {
                        //we are responsible for this key, so we look if we have a value
                        if (_dictionary.ContainsKey(i))
                        {
                            byte[] value = _dictionary[i].data;
                            if (sender == _overlay.LocalPeer.ID && RetrieveSucceeded != null)
                            {
                                RetrieveSucceeded.Invoke(this, new RetrieveEventArgs(key, value));
                            }
                            else if (sender != _overlay.LocalPeer.ID)
                            {
                                //send to another peer
                                var retrieveResponseMessage = new RetrieveResponseMessage();
                                retrieveResponseMessage.MessageType = MessageTypes.RETRIEVE_RESPONSE_MESSAGE;
                                retrieveResponseMessage.SenderID = _overlay.LocalPeer.ID;
                                retrieveResponseMessage.RecipientID = retrieveMessage.SenderID;
                                retrieveResponseMessage.Key = key;
                                retrieveResponseMessage.Value = value;
                                _overlay.Send(retrieveResponseMessage);
                            }
                        }
                        else
                        {
                            if (sender == _overlay.LocalPeer.ID && KeyNotFound != null)
                            {
                                KeyNotFound.Invoke(this, new RetrieveEventArgs(key, null));
                            }
                            else if (sender != _overlay.LocalPeer.ID)
                            {
                                var notFoundMessage = new NotFoundMessage();
                                notFoundMessage.MessageType = MessageTypes.NOT_FOUND_MESSAGE;
                                notFoundMessage.SenderID = _overlay.LocalPeer.ID;
                                notFoundMessage.RecipientID = retrieveMessage.SenderID;
                                notFoundMessage.Key = key;
                                _overlay.Send(notFoundMessage);
                            }
                        }
                    }
                    else
                    {
                        //we are NOT responsible for this key, so we route it to another peer
                        _overlay.Send(retrieveMessage);
                    }
                }
                else
                {
                    if (i > _startIndex || i <= _endIndex)
                    {
                        //we are responsible for this key, so we look if we have a value
                        if (_dictionary.ContainsKey(i))
                        {
                            byte[] value = _dictionary[i].data;;
                            if (sender == _overlay.LocalPeer.ID && RetrieveSucceeded != null)
                            {
                                RetrieveSucceeded.Invoke(this, new RetrieveEventArgs(key, value));
                            }
                            else if (sender != _overlay.LocalPeer.ID)
                            {
                                //send to another peer
                                var retrieveResponseMessage = new RetrieveResponseMessage();
                                retrieveResponseMessage.MessageType = MessageTypes.RETRIEVE_RESPONSE_MESSAGE;
                                retrieveResponseMessage.SenderID = _overlay.LocalPeer.ID;
                                retrieveResponseMessage.RecipientID = retrieveMessage.SenderID;
                                retrieveResponseMessage.Key = key;
                                retrieveResponseMessage.Value = value;
                                _overlay.Send(retrieveResponseMessage);
                            }
                        }
                        else
                        {
                            if (sender == _overlay.LocalPeer.ID && KeyNotFound != null)
                            {
                                KeyNotFound.Invoke(this, new RetrieveEventArgs(key, null));
                            }
                            else if (sender != _overlay.LocalPeer.ID)
                            {
                                //send to another peer  
                                var notFoundMessage = new NotFoundMessage();
                                notFoundMessage.MessageType = MessageTypes.NOT_FOUND_MESSAGE;
                                notFoundMessage.SenderID = _overlay.LocalPeer.ID;
                                notFoundMessage.RecipientID = retrieveMessage.SenderID;
                                notFoundMessage.Key = key;
                                _overlay.Send(notFoundMessage);
                            }
                        }
                    }
                    else
                    {
                        //we are NOT responsible for this key, so we route it to another peer
                        _overlay.Send(retrieveMessage);
                    }
                }
            }
            catch (Exception ex)
            {
                logger.Error("Exception during distributed Retrieve.", ex);
            }
        }

        public event EventHandler<StoreEventArgs> StoreSucceeded;
        public event EventHandler<StoreEventArgs> StoreFailed;
        public event EventHandler<RetrieveEventArgs> RetrieveSucceeded;
        public event EventHandler<RetrieveEventArgs> RetrieveFailed;
        public event EventHandler<RetrieveEventArgs> KeyNotFound;

        #endregion

        #region IDHT Members

        public int Size
        {
            get { return _dictionary.Count; }
        }

        #endregion
    }

    public class DHTEntry
    {
        public DateTime Timestamp { get; set; }
        public byte[] data;
    }
}
